using System;
using System.Collections.Generic;
using Me.Prettyprint.Cassandra.Model.Thrift;
using Me.Prettyprint.Cassandra.Serializers;
using Me.Prettyprint.Cassandra.Service;
using Me.Prettyprint.Cassandra.Utils;
using Me.Prettyprint.Hector.Api;
using Me.Prettyprint.Hector.Api.Beans;
using Me.Prettyprint.Hector.Api.Exceptions;
using Me.Prettyprint.Hector.Api.Mutation;
using Org.Apache.Cassandra.Thrift;

namespace Me.Prettyprint.Cassandra.Model
{
    /// <summary>
    /// Default implementation of Mutator interface. Mutators have to know the keyspace
    /// and key serializer to do it's job. If the user doesn't know the type of the
    /// key, the class uses TypeInferringSerializer to figure it out.
    /// </summary>
    /// <typeparam name="K"></typeparam>
	public sealed class MutatorImpl<K> : Mutator<K>
	{
        public MutatorImpl(Keyspace keyspace, Serializer<K> keySerializer, BatchSizeHint sizeHint)
        {
            Keyspace = (ExecutingKeyspace)keyspace;
            KeySerializer = keySerializer;
            SizeHint = sizeHint;
        }
        public MutatorImpl(Keyspace keyspace, Serializer<K> keySerializer) : this(keyspace, keySerializer, null)
        {
        }
        public MutatorImpl(Keyspace keyspace) : this(keyspace, TypeInferringSerializer<K>.Instance)
        {
        }
		public ExecutingKeyspace Keyspace
        {
            get;set;
        }
        public Serializer<K> KeySerializer
        {
            get;set;
        }
        public BatchMutation<K> PendingMutations
        {
            get;
            private set;
        }
        public BatchSizeHint SizeHint
        {
            get;
            private set;
        }
        public int GetPendingMutationCount()
        {
            return GetPendingMutations().Count;
        }

        private BatchMutation<K> GetPendingMutations()
        {
            if (PendingMutations == null)
            {
                PendingMutations = new BatchMutation<K>(KeySerializer, SizeHint);
            }
            return PendingMutations;
        }
        public Mutator<K> DiscardPendingMutations()
        {
            PendingMutations = null;
            return this;
        }
        public MutationResult Insert<N,V>(K key, string cf, HColumn<N,V> c)
        {
            AddInsertion(key, cf, c);
            return Execute();
        }
        public MutationResult Insert<SN,N,V>(K key, string cf, HSuperColumn<SN,N,V> superColumn)
        {
            AddInsertion(key, cf, superColumn);
            return Execute();
        }
        public MutationResult Delete<N>(K key, string cf, N columnName, Serializer<N> nameSerializer)
        {
            AddDeletion(key, cf, columnName, nameSerializer);
            return Execute();
        }
        public MutationResult Delete<N>(K key, string cf, N columnName, Serializer<N> nameSerializer, long clock) 
        {
            AddDeletion(key, cf, columnName, nameSerializer, clock);
            return Execute();
        }
        public Mutator<K> AddInsertion<N, V>(K key, string cf, HColumn<N, V> c)
        {
            GetPendingMutations().AddInsertion(key, cf, ((HColumnImpl<N, V>) c).ToThrift());
            return this;
        }
        public Mutator<K> AddInsertion<SN, N, V>(K key, string cf, HSuperColumn<SN, N, V> sc)
        {
            GetPendingMutations().AddSuperInsertion(key, Arrays.AsList(cf), ((HSuperColumnImpl<SN,N,V>) sc).ToThrift());
            return this;
        }
        public Mutator<K> AddDeletion<N>(K key, string cf, N columnName, Serializer<N> nameSerializer)
        {
            AddDeletion(key, cf, columnName, nameSerializer, Keyspace.CreateClock());
            return this;
        }
        public Mutator<K> AddDeletion<N>(K key, string cf)
        {
            AddDeletion(key, cf, default(N), null, Keyspace.CreateClock());
            return this;
        }
        public Mutator<K> AddDeletion<N>(K key, string cf, long clock)
        {
            AddDeletion(key, cf, default(N), null, clock);
            return this;
        }
        public Mutator<K> AddDeletion<N>(IEnumerable<K> keys, string cf)
        {
            return AddDeletion<N>(keys, cf, Keyspace.CreateClock());
        }
        public Mutator<K> AddDeletion<N>(IEnumerable<K> keys, string cf, long clock)
        {
            foreach (K key in keys)
            {
                AddDeletion(key, cf, default(N), null, clock);
            }
            return this;
        }
        public MutationResult InsertCounter<N>(K key, string cf, HCounterColumn<N> c)
        {
            AddCounter(key,cf, c);
            return Execute();
        }
        public MutationResult IncrementCounter<N>(K key, string cf, N columnName, long increment)
        {
            return InsertCounter(key, cf, new HCounterColumnImpl<N>(columnName, increment, TypeInferringSerializer<N>.Instance));
        }
        public MutationResult DecrementCounter<N>(K key, string cf, N columnName, long increment)
        {
            return IncrementCounter(key, cf, columnName, increment * -1L);
        }
        public MutationResult InsertCounter<SN, N>(K key, string cf, HCounterSuperColumn<SN, N> superColumn)
        {
            AddCounter(key, cf, superColumn);
            return Execute();
        }
        public MutationResult DeleteCounter<N>(K key, string cf, 
            N counterColumnName, Serializer<N> nameSerializer)
        {
            AddCounterDeletion(key,cf,counterColumnName,nameSerializer);
            return Execute();
        }
        public MutationResult SubDeleteCounter<SN, N>(K key, string cf, SN supercolumnName, 
            N columnName, Serializer<SN> sNameSerializer, Serializer<N> nameSerializer)
        {
            HCounterSuperColumnImpl<SN,N> superColumn = new HCounterSuperColumnImpl<SN,N>(sNameSerializer,nameSerializer);
            superColumn.Name = supercolumnName;
            superColumn.AddSubCounterColumn(new HCounterColumnImpl<N>(nameSerializer));
            AddCounterSubDeletion(key, cf, superColumn);
            return Execute();
        }
        public Mutator<K> AddCounter<N>(K key, string cf, HCounterColumn<N> c)
        {
            GetPendingMutations().AddCounterInsertion(key, Arrays.AsList(cf), ((HCounterColumnImpl<N>) c).ToThrift());
            return this;
        }
        public Mutator<K> AddCounter<SN, N>(K key, string cf, HCounterSuperColumn<SN, N> sc)
        {
            GetPendingMutations().AddSuperCounterInsertion(key, Arrays.AsList(cf), ((HCounterSuperColumnImpl<SN, N>) sc).ToThrift());
            return this;
        }
        public Mutator<K> AddCounterDeletion<N>(K key, string cf, N counterColumnName, Serializer<N> nameSerializer)
        {
            Deletion d;
            if ( counterColumnName != null )
            {
                SlicePredicate sp = new SlicePredicate();
                sp.Column_names.Add(nameSerializer.ToBytes(counterColumnName));
                d = new Deletion();
                d.Predicate = sp;
            } 
            else
            {
                d = new Deletion();
            }
            GetPendingMutations().AddDeletion(key, Arrays.AsList(cf), d);
            return this;
        }
        public Mutator<K> AddDeletion<N>(K key, string cf, N columnName, Serializer<N> nameSerializer, long clock)
        {
            Deletion d;
            if ( columnName != null ) 
            {
                SlicePredicate sp = new SlicePredicate();
                sp.Column_names = new List<byte[]>();
                sp.Column_names.Add(nameSerializer.ToBytes(columnName));
                d = new Deletion();
                d.Timestamp = clock;
                d.Predicate = sp;
            }
            else
            {
                d = new Deletion();
                d.Timestamp = clock;
            }
            GetPendingMutations().AddDeletion(key, Arrays.AsList(cf), d);
            return this;
        }
        public Mutator<K> AddCounterDeletion(K key, string cf)
        {
            GetPendingMutations().AddDeletion(key, Arrays.AsList(cf), new Deletion());
            return this;
        }
        public Mutator<K> AddCounterSubDeletion<SN, N>(K key, string cf, HCounterSuperColumn<SN, N> sc)
        {
            Deletion d = new Deletion();
            if ( sc.Columns != null )
            {
                SlicePredicate pred = new SlicePredicate();
                foreach (HCounterColumn<N> col in sc.Columns)
                {
                    pred.Column_names.Add(col.NameSerializer.ToBytes(col.GetName()));
                }
                d.Predicate = pred;
            }
            d.Super_column = sc.NameBytes;
            GetPendingMutations().AddDeletion(key, Arrays.AsList(cf), d);
            return this;
        }
        public Mutator<K> AddSubDelete<SN, N, V>(K key, string cf, HSuperColumn<SN, N, V> sc)
        {
            return AddSubDelete(key, cf, sc, Keyspace.CreateClock());
        }
        public Mutator<K> AddSubDelete<SN, N, V>(K key, string cf, HSuperColumn<SN, N, V> sc, long clock)
        {
            Deletion d = new Deletion();
            d.Timestamp = clock;
            if ( sc.Columns != null )
            {
                SlicePredicate pred = new SlicePredicate();
                foreach (HColumn<N, V> col in sc.Columns)
                {
                    pred.Column_names.Add(col.NameSerializer.ToBytes(col.Name));
                }
                d.Predicate = pred;
            }
            d.Super_column = sc.NameBytes;
            List<string> cfs = new List<string>(cf.Split(','));
            GetPendingMutations().AddDeletion(key, cfs, d);        
            return this;
        }
        public Mutator<K> AddSubDelete<SN, N>(K key, string cf, SN sColumnName, 
            N columnName, Serializer<SN> sNameSerializer, Serializer<N> nameSerializer)
        {
            return AddSubDelete(key, cf, sColumnName, columnName, sNameSerializer, nameSerializer, Keyspace.CreateClock());
        }
        public Mutator<K> AddSubDelete<SN, N>(K key, string cf, SN sColumnName,
            N columnName, Serializer<SN> sNameSerializer, Serializer<N> nameSerializer, long clock)
        {
            Deletion d = new Deletion();
            d.Timestamp = clock;
            SlicePredicate predicate = new SlicePredicate();
            predicate.Column_names.Add(nameSerializer.ToBytes(columnName));
            d.Predicate = predicate;
            d.Super_column = sNameSerializer.ToBytes(sColumnName);
            GetPendingMutations().AddDeletion(key, Arrays.AsList(cf), d);
            return this;
        }
        public Mutator<K> AddSuperDelete<SN>(K key, string cf, SN sColumnName, Serializer<SN> sNameSerializer)
        {
            Deletion d = new Deletion();
            d.Timestamp = Keyspace.CreateClock();
            d.Super_column = sNameSerializer.ToBytes(sColumnName);
            GetPendingMutations().AddDeletion(key, Arrays.AsList(cf), d);
            return this;
        }
        public MutationResult Execute()
        {
            if (PendingMutations == null || PendingMutations.Count == 0) 
            {
                return new MutationResultImpl<object>(true, 0, null);
            }
            BatchMutation<K> mutations = PendingMutations.makeCopy();
            PendingMutations = null;
            BatchMutateOperation<K> operation = new BatchMutateOperation<K>(OperationType.WRITE, mutations);
            ExecutionResult<object> execRes = Keyspace.DoExecuteOperation<object>(operation);
            MutationResult result = new MutationResultImpl<object>(execRes);
            return result;
        }
        public MutationResult SubDelete<SN,N>(K key, string cf, SN supercolumnName,
            N columnName, Serializer<SN> sNameSerializer, Serializer<N> nameSerializer)
        {
            Func<KeyspaceService, object> x = (KeyspaceService ks) =>
            {
                ks.Remove(KeySerializer.ToBytes(key),
                    ThriftFactory.CreateSuperColumnPath(cf, supercolumnName, columnName, sNameSerializer, nameSerializer));
                return null;
            };
            ExecutionResult<object> result = Keyspace.DoExecute(new KeyspaceOperationCallbackBuilder<object>(x));
            return new MutationResultImpl<object>(result);
        }
        public MutationResult SuperDelete<SN>(K key, string cf, 
            SN supercolumnName, Serializer<SN> sNameSerializer)
        {
            Func<KeyspaceService, object> x = (KeyspaceService ks) =>
            {
                ks.Remove(KeySerializer.ToBytes(key), ThriftFactory.CreateSuperColumnPath(cf, supercolumnName, sNameSerializer));
                return null;
            };
            ExecutionResult<object> result = Keyspace.DoExecute(new KeyspaceOperationCallbackBuilder<object>(x));
            return new MutationResultImpl<object>(result);
        }
	}
}